use std::{collections::{hash_map::Entry, HashMap, HashSet}, path::{Path, PathBuf}};

use serde::{Deserialize, Serialize};
use wtask_app::{AppCmd, AppFileDownload, AppFileUpload, AppTaskSubmit, CmdArgs, CmdRet, FileUDArgs, FuturesUnordered, StreamExt, TaskSubmitMeta, WPath, WTaskSubmitMetaBase, cli_app, msg_func_none};
use wtask_base::{error::{WError, WResult}, export::{JoinError, JoinHandle, log_debug, tokio_spawn}, logs::log_format, machine::CMDArgsConfigMachine};

use crate::{basic::{MachineConfigs, WEB_SESSION}, handle::flow::basic::{FlowData, FlowEdgeData, FlowNodeData, FlowNodeDataMeta, FlowShellMeta, FlowStatus, FlowStatusNodeData, FlowStatusTaskData, FlowTaskMeta}};




impl FlowStatus {
    fn push_log(&mut self, log_data: String) {
        self.log.push(log_format(log_data));
    }

    fn task_add(&mut self, node_id: &str, data: FlowStatusTaskAdd) -> WResult<()> {
        self.task.get_mut(node_id)
            .ok_or(WError::Empty)?
            .dir_result
            .insert(data.machine_id, data.dir_result);
        Ok(())
    }
    
    fn task_new(&mut self, node_id: String, task_id: String, data: FlowStatusTaskAdd) {
        let data = {
            let mut dir_result = HashMap::new();
            dir_result.insert(data.machine_id, data.dir_result);
            FlowStatusTaskData {
                task_id,
                dir_result
            }
        };
        self.task.insert(node_id, data);
    }

    fn task_get(&self, node_id: &str, machine_id: &str) -> Option<&Path> {
        self.task
            .get(node_id)
            .and_then(| v | v.dir_result.get(machine_id))
            .map(| v | v.as_path())
    }

    fn done_edge_status(&mut self, log_data: String, edge_id: String, status: bool) {
        self.running.remove(&edge_id);
        self.edge.insert(edge_id, status);
        self.push_log(log_data);
    }


    fn done_node_status(&mut self, node_id: String, node_result: Option<FlowStatusNodeData>, log_data: Vec<String>) {
        self.running.remove(&node_id);
        self.log.extend(log_data);
        self.node.insert(node_id, node_result);
    }
}







#[derive(Debug, Clone, Serialize, Deserialize)]
struct FlowNodeResultTask {
    task_id: String,
    machine_id: String,
    dir_result: PathBuf,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
enum FlowNodeResult {
    // 直接存储数据
    Data(String),
    // 任务数据
    Task(FlowNodeResultTask),
}


impl From<&FlowNodeResult> for FlowStatusNodeData {
    fn from(value: &FlowNodeResult) -> Self {
        match value {
            FlowNodeResult::Data(v) => Self::Data(v.to_string()),
            FlowNodeResult::Task(v) => Self::Task(v.machine_id.clone())
        }
    }
}




struct FlowNodeRes {
    success: bool,
    node_result: FlowNodeResult,
    logs: Vec<String>,
}

impl FlowNodeRes {
    fn cmdret_to_noderes(machine_id: String, data: CmdRet, logs: Vec<String>, res_last_log: bool) -> Self {
        let node_result = if res_last_log {
            let v = if logs.len() > 1 {
                let mut inp_next = logs.get(logs.len() - 2).cloned().unwrap_or(String::new());
                log_debug!("node shell res last log: {:?}", inp_next);
                if let Some(i) = inp_next.find("]") {
                    inp_next.split_off(i+1)
                } else {
                    inp_next
                }
            } else {
                String::new()
            };
            FlowNodeResult::Data(v)
        } else {
            FlowNodeResult::Task(FlowNodeResultTask {
                task_id: data.id,
                machine_id,
                dir_result: data.dir_result
            })
        };
        if let Some(res) = logs.last() && res.trim_end().ends_with(" success") {
            FlowNodeRes {
                success: true,
                node_result,
                logs: Vec::with_capacity(0),
            }
        } else {
            FlowNodeRes {
                success: false,
                node_result,
                logs,
            }
        }
    }
}




struct FlowNodeHandle {
    node_id: String,
    node_info: String,
    res: WResult<FlowNodeRes>,
}


struct FlowEdgeNodeData {
    machine_id: String,
    node_data: FlowNodeDataMeta,
    machine_config: CMDArgsConfigMachine,
    node_note: String,
    node_info: String,
    // $node_id$ -> inp
    node_inps: HashMap<String, String>,
}

struct FlowStatusTaskAdd {
    machine_id: String,
    dir_result: PathBuf,
}


struct FlowEdgeHandle {
    edge_id: String,
    node_id_now: String,
    node_inp_id: String,
    res: WResult<FlowStatusTaskAdd>,
}



// edge child/parent: node_id -> Vec<node_id>
pub(super) type EdgeRelated = HashMap<String, HashSet<String>>;
// edge id
pub(super) type EdgeId = HashMap<String, HashMap<String, String>>;

pub(super) struct FlowGraphEdges {
    // 关系: node_id -> Vec<node_id>
    pub(super) edges_child: EdgeRelated,
    pub(super) edges_parent: EdgeRelated,
    // 关系：node_id_parent -> {node_id_child -> edge_id}
    pub(super) edges_id: EdgeId,
}

pub(super) struct FlowGraph {
    // meta
    flow_id: String,
    flow_name: String,
    // 运行的设备: machine_token -> machine_config
    machines: MachineConfigs,
    // 所有节点: node_id -> (node_name, node_meta)
    nodes: HashMap<String, (String, FlowNodeDataMeta)>,
    // 边关系
    edges: FlowGraphEdges,
    // 状态
    status: FlowStatus,
    // 运行edge时，临时存储的等待运行的node: node_id -> node_next
    node_next: HashMap<String, FlowEdgeNodeData>,
    // 文件中转的handle
    handles_edge: FuturesUnordered<JoinHandle<FlowEdgeHandle>>,
    // 运行任务的handle
    handles_node: FuturesUnordered<JoinHandle<FlowNodeHandle>>,
}

impl Drop for FlowGraph {
    fn drop(&mut self) {
        for handle in self.handles_edge.iter() {
            handle.abort();
        }
        for handle in self.handles_node.iter() {
            handle.abort();
        }
    }
}

impl FlowGraph {
    /// 提供所有nodes
    pub(super) fn find_start_node<T>(edges_child: &EdgeRelated, nodes: &HashMap<String, T>) -> WResult<HashSet<String>> {
        let nodes_id_next = edges_child.values().flatten().collect::<HashSet<_>>();
        let nodes_id_start = nodes.keys()
            .filter(| &v | !nodes_id_next.contains(v))
            .cloned()
            .collect::<HashSet<_>>();
        Ok(nodes_id_start)
    }

    /// 节点完成后找到下一些节点进行计算
    /// 1. 找到该节点的所有子节点
    /// 2. 筛选子节点，要求子节点其所有父节点均已完成
    /// 
    /// 提供已经完成的nodes
    pub(super) fn find_next_node<T>(node_id: &str, edges_child: &EdgeRelated, edges_parent: &EdgeRelated, nodes: &HashMap<String, T>) -> WResult<HashSet<String>> {
        let nodes_id_next = edges_child
            .get(node_id)
            .map(| ids | {
                ids.iter()
                    .filter(| &v | {
                        if let Some(nodes_prev) = edges_parent.get(v) {
                            nodes_prev.iter().all(| vv | nodes.contains_key(vv))
                        } else {
                            true
                        }
                    })
                    .cloned()
                    .collect::<HashSet<_>>()
            })
            .unwrap_or_default();
        Ok(nodes_id_next)
    }

    #[allow(clippy::type_complexity)]
    pub(super) fn build_edges(edges: HashMap<String, FlowEdgeData>) -> FlowGraphEdges {
        let mut edges_child: EdgeRelated = HashMap::new();
        let mut edges_parent: EdgeRelated = HashMap::new();
        let mut edges_id: EdgeId = HashMap::new();
        let f = |edges_temp: &mut EdgeRelated, edge_k: String, edge_v: String| {
            match edges_temp.entry(edge_k) {
                Entry::Occupied(mut v) => { v.get_mut().insert(edge_v); },
                Entry::Vacant(v) => {
                    let mut a = HashSet::new();
                    a.insert(edge_v);
                    v.insert(a);
                },
            }
        };
        for (edge_id, edge_temp) in edges.into_iter() {
            let edge_temp2 = edge_temp.clone();
            let edge_temp3 = edge_temp.clone();
            match edges_id.entry(edge_temp3.from) {
                Entry::Occupied(mut v) => { v.get_mut().insert(edge_temp3.to, edge_id); },
                Entry::Vacant(v) => {
                    let mut a = HashMap::new();
                    a.insert(edge_temp3.to, edge_id);
                    v.insert(a);
                },
            }
            f(&mut edges_child, edge_temp.from, edge_temp.to);
            f(&mut edges_parent, edge_temp2.to, edge_temp2.from);
        }
        FlowGraphEdges {
            edges_child,
            edges_parent,
            edges_id
        }
    }


    fn get_node_machines<'b>(&'b self, node_id: &str) -> WResult<HashSet<&'b str>> {
        if let Some(m) = self.status.task.get(node_id) {
            let mut a = HashSet::with_capacity(m.dir_result.len());
            a.extend(m.dir_result.keys().map(| v | v.as_str()));
            return Ok(a);
        }
        let node_data = self.nodes.get(node_id).ok_or(WError::Empty)?;
        let m = match &node_data.1 {
            FlowNodeDataMeta::Task(task_temp) => task_temp.machines.iter().map(| v | v.as_str()).collect::<HashSet<_>>(),
            FlowNodeDataMeta::Shell(_) => self.machines.keys().map(| v | v.as_str()).collect::<HashSet<_>>(),
            FlowNodeDataMeta::Arg(_) => HashSet::with_capacity(0)
        };
        Ok(m)
    }

    fn get_node_machines_related<'b>(&'b self, nodes_related: &'b EdgeRelated, node_id: &str) -> WResult<Vec<&'b str>> {
        let r = if let Some(nodes_related) = nodes_related.get(node_id) {
            nodes_related.iter()
                .map(| v | self.get_node_machines(v))
                .collect::<WResult<Vec<_>>>()?
                .into_iter()
                .flatten()
                .collect::<Vec<_>>()
        } else {
            Vec::with_capacity(0)
        };
        Ok(r)
    }

    /// 将prev和next合并，找出now里面最多的
    fn select_machine(&self, node_id: &str) -> WResult<CMDArgsConfigMachine> {
        let machine_now = self.get_node_machines(node_id)?;
        let machine_prev = self.get_node_machines_related(&self.edges.edges_parent, node_id)?;
        let machine_next = self.get_node_machines_related(&self.edges.edges_child, node_id)?;
        // count
        // TODO: 权重：设备状态、设备是子/父节点
        let mut machine_count = HashMap::with_capacity(machine_now.len());
        for machine_id in machine_prev.into_iter().chain(machine_next.into_iter()) {
            if !machine_now.contains(machine_id) {
                continue;
            }
            match machine_count.entry(machine_id) {
                Entry::Occupied(mut v) => { *v.get_mut() += 1; },
                Entry::Vacant(v) => { v.insert(1u8); },
            }
        }
        // max
        let machine_max = if machine_count.is_empty() {
            machine_now.into_iter().next().ok_or(WError::DataError("node no machine".to_owned()))?
        } else {
            machine_count
                .into_iter()
                .max_by_key(| v | v.1)
                .ok_or(WError::DataError("machine count max error".to_owned()))?
                .0
        };
        self.machines
            .get(machine_max)
            .cloned()
            .ok_or(WError::DataError("machine config get error".to_owned()))
    }
    
    fn run_node_arg(node_args: Option<String>, arg_meta: WTaskSubmitMetaBase) -> WResult<FlowNodeRes> {
        if let Some(v) = node_args {
            let errs = arg_meta.check(&v)?;
            if errs.is_empty() {
                Ok(FlowNodeRes {
                    success: true,
                    node_result: FlowNodeResult::Data(v),
                    logs: vec![log_format("Check success".to_owned())],
                })
            } else {
                Err(WError::RunError(errs.into_iter().map(log_format).collect::<Vec<_>>()))
            }
        } else if arg_meta.required {
            Err(WError::RunError(vec![log_format("No input found".to_owned())]))
        } else {
            Ok(FlowNodeRes {
                success: true,
                node_result: FlowNodeResult::Data(String::new()),
                logs: vec![log_format("No input for check".to_owned())],
            })
        }
    }

    fn inp_replace(mut inp: String, inps: &HashMap<String, String>) -> String {
        for (k, v) in inps {
            inp = inp.replace(k, v);
        }
        inp
    }
    
    async fn run_node_task(machine_config: &CMDArgsConfigMachine, task_meta: FlowTaskMeta, inps: HashMap<String, String>, machine_id: String, node_note: String) -> WResult<FlowNodeRes> {
        let args_inp = task_meta.args
            .get(&machine_id)
            .ok_or(WError::DataError("arg inp get machine error".to_owned()))?
            .iter()
            .map(| (k, v) | format!("{}={}", k, Self::inp_replace(v.to_owned(), &inps)))
            .collect::<Vec<_>>();
        let app = AppTaskSubmit::from(TaskSubmitMeta {
            app: task_meta.id,
            version: task_meta.version,
            dir_result: None,
            tag: "编排预定义任务".to_owned(),
            note: node_note,
            args: args_inp
        });
        let mut logs = Vec::new();
        let data = cli_app(app, machine_config, None, |_, v| logs.push(v))
            .await?
            .ok_or(WError::Empty)?;
        let res = FlowNodeRes::cmdret_to_noderes(machine_id, data, logs, false);
        Ok(res)
    }

    
    async fn run_node_shell(machine_config: &CMDArgsConfigMachine, shell_meta: FlowShellMeta, inps: HashMap<String, String>, machine_id: String, node_note: String) -> WResult<FlowNodeRes> {
        let app = AppCmd::from(CmdArgs {
            note: node_note,
            dir_result: None,
            image: "local".to_owned(),
            tag: "编排Shell任务".to_owned(),
            envs: Vec::new(),
            cmd: vec![Self::inp_replace(shell_meta.cmd, &inps)],
        });
        let mut logs = Vec::new();
        let data = cli_app(app, machine_config, None, |_, v| logs.push(v))
            .await?
            .ok_or(WError::Empty)?;
        let res = FlowNodeRes::cmdret_to_noderes(machine_id, data, logs, shell_meta.res_last_log);
        Ok(res)
    }

    fn done_node(&mut self, node_handle: FlowNodeHandle) -> WResult<()> {
        match node_handle {
            FlowNodeHandle { node_id, node_info, res: Ok(node_res)} if node_res.success => {
                // node完成
                let mut log_data = node_res.logs;
                log_data.push(log_format(format!("Node {} done", node_info)));
                self.status.done_node_status(node_id.clone(), Some((&node_res.node_result).into()), log_data);
                if let FlowNodeResult::Task(task_res) = node_res.node_result {
                    self.status.task_new(
                        node_id.clone(),
                        task_res.task_id,
                        FlowStatusTaskAdd {
                            machine_id: task_res.machine_id,
                            dir_result: task_res.dir_result
                        }
                    );
                }
                let nodes_id_next = Self::find_next_node(
                    &node_id,
                    &self.edges.edges_child,
                    &self.edges.edges_parent,
                    &self.status.node
                )?;
                log_debug!("find next node: {} {} {} {:?}", self.flow_id, self.flow_name, node_id, nodes_id_next);   
                self.run_nodes(nodes_id_next)?;
                Ok(())
            },
            FlowNodeHandle { node_id, node_info, res: Ok(node_res)} => {
                // node失败
                let mut log_data = node_res.logs;
                log_data.push(log_format(format!("Node {} error", node_info)));
                self.status.done_node_status(node_id, None, log_data);
                Err(WError::Stop)
            },
            FlowNodeHandle { node_id, node_info, res: Err(e) } => {
                // node失败
                let log_data = match e {
                    WError::RunError(mut log_data) => {
                        log_data.push(log_format(format!("Node {} error", node_info)));
                        log_data
                    },
                    ee => vec![log_format(ee.to_string())]
                };
                self.status.done_node_status(node_id, None, log_data);
                Err(WError::Stop)
            },
        }
    }


    fn done_edge(&mut self, edge_handle: FlowEdgeHandle) -> WResult<()> {
        match edge_handle {
            FlowEdgeHandle { edge_id, node_id_now, node_inp_id, res: Ok(res) } => {
                // edge完成
                self.node_next.get_mut(&node_id_now)
                    .ok_or(WError::Empty)?
                    .node_inps
                    .insert(node_inp_id, res.dir_result.to_string_lossy().to_string());
                self.status.done_edge_status(format!("Edge {} done", edge_id), edge_id, true);
                self.status.task_add(&node_id_now, res)?;
                Ok(())
            },
            FlowEdgeHandle { edge_id, node_id_now: _, node_inp_id: _, res: Err(e) } => {
                // edge传输失败
                self.status.done_edge_status(format!("Edge {} error: {}", edge_id, e), edge_id, false);
                Err(WError::DataError("Edge run error".to_owned()))
            },
        }
    }


    fn done_node_or_edge<T, F: FnMut(T)->WResult<()>>(handle: Option<Result<T, JoinError>>, mut handle_func: F) -> WResult<bool> {
        let no_next = match handle {
            Some(Ok(handle_res)) => {
                handle_func(handle_res)?;
                false
            },
            Some(Err(e_join)) => {
                // node启动失败
                return Err(WError::DataError(format!("Start error: {}", e_join)));
            },
            _ => {
                log_debug!("No task left, exit");
                true
            },
        };
        Ok(no_next)
    }

    fn start_node(&mut self, node_id: String, node_next: FlowEdgeNodeData) -> WResult<()> {
        self.status.running.insert(node_id.clone());
        self.status.push_log(format!("Node {} start", node_next.node_info));
        let h = tokio_spawn(async move {
            let res = match node_next.node_data {
                FlowNodeDataMeta::Task(task_meta) => Self::run_node_task(
                    &node_next.machine_config,
                    task_meta,
                    node_next.node_inps,
                    node_next.machine_id,
                    node_next.node_note
                ).await,
                FlowNodeDataMeta::Shell(shell) => Self::run_node_shell(
                    &node_next.machine_config,
                    shell,
                    node_next.node_inps,
                    node_next.machine_id,
                    node_next.node_note
                ).await,
                _ => Err(WError::DataError("Arg error".to_owned())),
            };
            FlowNodeHandle {
                node_id,
                node_info: node_next.node_info,
                res
            }
        });
        self.handles_node.push(h);
        Ok(())
    }


    fn edge_transfer(&self, edge_id: String, node_id_now: String, node_inp_id: String, machine_now: CMDArgsConfigMachine, machine_prev: CMDArgsConfigMachine, dir_result_prev: PathBuf) -> WResult<JoinHandle<FlowEdgeHandle>> {
        // diff machine, need transfer file
        // from machine_prev dir_result_prev
        // to machine_now

        // start edge
        let h = tokio_spawn(async move {
            let res_func = async move || {
                // download
                let mut app_download = AppFileDownload::from(FileUDArgs::new(dir_result_prev));
                app_download.zip_dir = true;
                let app_download_res = cli_app(app_download, &machine_prev, None, msg_func_none)
                    .await?
                    .ok_or(WError::Empty)?;
                // path
                let file_upload = WPath::try_from(app_download_res)?.into();
                // upload
                let mut app_upload = AppFileUpload::from(FileUDArgs::new(file_upload));
                app_upload.dir = "result".to_owned();
                app_upload.unzip = true;
                let app_upload_res = cli_app(app_upload, &machine_now, None, msg_func_none)
                    .await?
                    .ok_or(WError::Empty)?;
                // res
                Ok(FlowStatusTaskAdd {
                    machine_id: machine_now.machine_token,
                    dir_result: app_upload_res.into()
                })
            };
            let res: WResult<FlowStatusTaskAdd> = res_func().await;
            FlowEdgeHandle {
                edge_id,
                node_id_now,
                node_inp_id,
                res
            }
        });
        Ok(h)
    }

    // return edge_data, or transfer data
    fn edge_data(&self, node_id_prev: &str, machine_now: &str, node_prev_result: FlowStatusNodeData) -> (Option<String>, Option<String>) {
        match (self.status.task_get(node_id_prev, machine_now), node_prev_result) {
            // has transfer
            (Some(v), _) => (Some(v.to_string_lossy().to_string()), None),
            // data, no need transfer
            (None, FlowStatusNodeData::Data(v)) => (Some(v), None),
            // task, need transfer
            (None, FlowStatusNodeData::Task(machine_id_prev)) => (None, Some(machine_id_prev)),
        }
    }



    fn set_nodes_next(&mut self, node_id: String, mut node_next: FlowEdgeNodeData) -> WResult<()> {
        let node_prev = self.edges.edges_parent.get(&node_id);
        if node_prev.is_none() {
            self.node_next.insert(node_id, node_next);
            return Ok(());
        }
        
        let nodes_next_raw = node_prev.unwrap()
            .iter()
            .filter_map(| node_prev_temp | self.status.node
                .get(node_prev_temp)
                .cloned()
                .unwrap_or_default()
                .map(| v | (format!("${}$", node_prev_temp), node_prev_temp, v)))
            .collect::<Vec<_>>();
        
        
        for (node_inp_id, node_id_prev, node_prev_result) in nodes_next_raw {
            let edge_id = self.edges.edges_id
                .get(node_id_prev)
                .and_then(| v | v.get(&node_id).cloned())
                .ok_or(WError::Empty)?;
            match self.edge_data(node_id_prev, &node_next.machine_id, node_prev_result) {
                (Some(node_inp_temp), _) => {
                    // no need transfer
                    node_next.node_inps.insert(node_inp_id, node_inp_temp);
                    self.status.done_edge_status(format!("Edge {} done", edge_id), edge_id, true);
                },
                (_, Some(machine_prev)) => {
                    // need transfer
                    let dir_result_prev = self.status
                        .task_get(node_id_prev, &machine_prev)
                        .ok_or(WError::Empty)?
                        .to_path_buf();
                    let machine_config_prev = self.machines
                        .get(&machine_prev)
                        .ok_or(WError::Empty)?
                        .clone();
                    let machine_config_now = self.machines
                        .get(&node_next.machine_id)
                        .ok_or(WError::Empty)?
                        .clone();
                    let handle = self.edge_transfer(
                        edge_id,
                        node_id.clone(),
                        node_inp_id,
                        machine_config_now,
                        machine_config_prev,
                        dir_result_prev
                    )?;
                    self.handles_edge.push(handle);
                },
                _ => {
                    return Err(WError::DataError("Prev machine not run".to_owned()))
                }
            }
        }
        self.node_next.insert(node_id, node_next);
        Ok(())

    }

    fn run_nodes(&mut self, nodes_id: HashSet<String>) -> WResult<()> {
        for node_id in nodes_id {
            // data: node_data, node_args, machine
            let (node_name, node_data) = self.nodes
                .get(&node_id)
                .cloned()
                .ok_or(WError::DataError("node get error".to_owned()))?;
            let node_info = format!("{}({})", node_name, node_id);
            
            let node_data = match node_data {
                FlowNodeDataMeta::Arg(arg_meta) => {
                    let machine_id = self.machines.keys().next().unwrap();
                    let node_args = self.status.submit.args
                        .get(machine_id)
                        .ok_or(WError::DataError("arg get error".to_owned()))?
                        .get(&node_id)
                        .cloned();
                    let node_handle = FlowNodeHandle {
                        node_id,
                        node_info,
                        res: Self::run_node_arg(node_args, arg_meta)
                    };
                    self.done_node(node_handle)?;
                    continue;
                },
                node_data => node_data
            };

            let machine_config = self.select_machine(&node_id)?;
            let machine_id = machine_config.machine_token.clone();
            let node_note = format!("【{}】{}", self.flow_name, node_info);
            let node_next = FlowEdgeNodeData {
                machine_id,
                node_data,
                machine_config,
                node_note,
                node_info,
                node_inps: HashMap::new(),
            };
            self.set_nodes_next(node_id, node_next)?;
        }
        Ok(())
    }


    pub(super) fn new(flow_data: FlowData, flow_status: FlowStatus, machines: MachineConfigs) -> WResult<Self> {
        let edges = Self::build_edges(flow_data.edges);
        let nodes = {
            let mut nodes_now = HashMap::new();
            for (node_id, node_data) in flow_data.nodes.into_iter() {
                let node_data_temp = FlowNodeData::parse_meta(
                    &node_data.cls,
                    node_data.args,
                    node_data.meta
                )?;
                nodes_now.insert(node_id, (node_data.name, node_data_temp));
            }
            nodes_now
        };
        Ok(Self {
            flow_id: flow_data.setting.id,
            flow_name: flow_data.setting.name,
            status: flow_status,
            edges,
            nodes,
            machines,
            node_next: HashMap::new(),
            handles_edge: FuturesUnordered::new(),
            handles_node: FuturesUnordered::new(),
        })
    }

    fn has_left_node(&self) -> bool {
        self.nodes.len() != self.status.node.len()
    }

    /// 运行flow
    /// 
    /// 整体的输入：flow_args, 根据node(cls not in (task, shell))的参数约束规则进行输入，因此首先需要检查输入的合法性
    /// 
    /// 连接关系：edge
    /// 
    /// 节点的内容：
    ///     1. node(cls=shell)，内容在meta内
    ///     2. node(cls=task)，内容在args内
    /// 
    /// 运行逻辑：
    ///     1. 根据edge连接关系构建一个运算图
    ///     2. 找到最开始的节点：可能多个
    ///     3. 从开始节点运行
    ///         
    ///         1. 多个同时运行
    ///         2. 当一个运行结束，根据运算图向下找节点并行运行（可能多个节点）
    ///         3. 运行节点前计算该节点运行的machine：选择machine需要一个单独的算法，可以随机，可以根据输入是否有文件及上下节点machine
    pub(super) async fn run(mut self) -> WResult<bool> {
        let nodes_id_start = Self::find_start_node(&self.edges.edges_child, &self.nodes)?;
        log_debug!("find start node: {} {} {:?}", self.flow_id, self.flow_name, nodes_id_start);
        self.run_nodes(nodes_id_start)?;
        let mut success = true;
        while self.has_left_node() {
            WEB_SESSION.flow.write().await.status_save(&self.status)?;
            // 先把edge运行完毕
            loop {
                let handle = self.handles_edge.next().await;
                match Self::done_node_or_edge(handle, | v | self.done_edge(v)) {
                    Ok(false) => {},
                    Err(e) => {
                        self.status.push_log(format!("Edge error: {}", e));
                        success = false;
                        break;
                    },
                    Ok(true) => {
                        break;
                    }
                }
            }
            if !success {
                break;
            }
            WEB_SESSION.flow.write().await.status_save(&self.status)?;
            // 然后运行node
            let nodes_next = self.node_next.drain().collect::<Vec<_>>();
            for (node_id, node_next) in nodes_next {
                self.start_node(node_id, node_next)?;
            }
            loop {
                let handle = self.handles_node.next().await;
                match Self::done_node_or_edge(handle, | v | self.done_node(v)) {
                    Ok(false) => {},
                    Err(e) => {
                        self.status.push_log(format!("Node error: {}", e));
                        success = false;
                        break;
                    },
                    Ok(true) => {
                        break;
                    }
                }
            }
            if !success {
                break;
            }
            log_debug!("wait next node done ...");
        }
        WEB_SESSION.flow.write().await.status_save(&self.status)?;
        Ok(success)
    }
}




